package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"log"
	"time"
)

func main() {
	AsyncProducer()
}


var addrs = []string{
	"b-3.prod-kafka-01.fotqj4.c3.kafka.ap-east-1.amazonaws.com:9092",
	"b-1.prod-kafka-01.fotqj4.c3.kafka.ap-east-1.amazonaws.com:9092",
	"b-2.prod-kafka-01.fotqj4.c3.kafka.ap-east-1.amazonaws.com:9092",
}

//var addrs = []string{"16.162.166.79:49153"}

var topic = "amazon_streamapi_sp_ods_traffic_qa"

func AsyncProducer() {

	producer, err := sarama.NewAsyncProducer(addrs, DefaultConfig())

	if err != nil {
		fmt.Printf(err.Error())
	}

	fmt.Println("producter connect")

	msg := &sarama.ProducerMessage{
		Topic: topic,
		Key:   nil,
	}

	for i := 0; i < 100; i++ {
		log.Println("cc", i)
		//msg.Value = sarama.ByteEncoder(fmt.Sprintf("gary_test_%d", i))
		//go func(msg *sarama.ProducerMessage, topic string, i int) {
			msg.Value = sarama.ByteEncoder(fmt.Sprintf("gary_test_%d", i))
			producer.Input() <- msg
			//fmt.Println("push_msg-", topic, i)

			select {
			case suc := <-producer.Successes():
				//log.Println(i)
				log.Println(msg.Value, suc.Offset, i)
			case fail := <-producer.Errors():
				//fmt.Println("失败")
				log.Println(fail.Error())

			case <-time.After(10 * time.Second):
				// 如果超时，则退出
				log.Println("超时退出")

			}
		//}(msg, topic, i)
	}

	time.Sleep(50 * time.Second)

}


func SyncProducer() {

}



func DefaultConfig() *sarama.Config {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForLocal
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V2_4_0_0
	return config
}


